/*
 * Name     : Worker.java
 * Author(s): Jill San Luis
 * Created  : Apr 6, 2011, 12:44:41 AM
 */

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.data.Stat;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.math.BigInteger;
/**
 * The Worker class
 * 
 * @author Jill San Luis
 */
public class Worker {
	
	/** Default Worker path */
	public static final String WORKER_PATH = PoolCreator.WORKERPOOL_PATH + "/worker"; 
    
	public static final String FILE_REQUEST_PATH = PoolCreator.FILEPOOL_PATH + "/filerequest";
	
	/** Thread sleep duration */
    private static final int SLEEP_TIME = 500;
   
	/** Manager for ZooKeeper server connection */
	private ZkConnector zkc;
	
	/** Path to this Worker zNode */
	private String zNodePath ;
	
	/** Watcher for this Worker zNode */
	private Watcher workerWatcher;
	
	/** Current password hash in process */
	private String currentHash;
	
	private String currentTaskFolder;
	
	private boolean fileServerActiveFlag = true;
	
	/**
	 * Creates a new instance of Worker.
	 * 
	 * @param zkServers	a comma separated list of hosts that can be connected to.
	 */
	private Worker(String zkServers) {

		zkc = new ZkConnector();

		try {
			zkc.connect(zkServers);
		} catch (Exception e) {
			System.out.println("Zookeeper connect " + e.getMessage());
		}

		workerWatcher = new Watcher(){	
			public void process(WatchedEvent event){
				handleUpdate(event);
			}
		};
		
		try{
			PoolCreator.createPool(zkc, PoolCreator.WORKERPOOL_PATH);
			zNodePath = zkc.getZooKeeper().create(WORKER_PATH, null, zkc.getACL(), CreateMode.EPHEMERAL_SEQUENTIAL);
		}
		catch(Exception e){		
			System.out.println("Error: EXITING... Cannot create Worker zNode - " + e.getMessage());
			System.exit(-1);
		}		
		
		// creating File Pool path if not already created 
		PoolCreator.createPool(zkc,PoolCreator.FILEPOOL_PATH);
	}
	
	/**
	 * 
	 * @param event
	 */
	private void handleUpdate(WatchedEvent event){
		
		if(event.getType() == EventType.NodeDataChanged){
			String content = null;
			
			try{
				Stat readByteStat = new Stat();
				byte[] readByte = zkc.getZooKeeper().getData(event.getPath(), true, readByteStat);
				
				content = new String(readByte);
				
				System.out.println("Info: Read [" + content + "] from " + event.getPath());
								
				if(content.isEmpty()){
					System.out.println("Error: The data received in "+ event.getPath() + " is empty!");
					return;
				}
			}
			catch(KeeperException ke){
				System.out.println("Error: Cannot read the data in " + event.getPath() + " " + ke.getMessage());
				return;
			}
			catch(InterruptedException ie){
				
				System.out.println("Error: Cannot read the data in " + event.getPath() + " " + ie.getMessage());
				return;
			}
			
			// parse the content of the data inside
			String[] token = content.split(ZkConnector.DATADELIMITER);

			System.out.println("Info: Received " + token.length + " tokens.");
			for(String t: token){
				System.out.println("===================");
				System.out.println(t);
			}
			
			// when invoked by JobTracker, there should only be one token in the 
			// data received - path to task. 
			if(token.length == 1){

				processWorkerFolderContent(content, token[0], event.getPath());
				
			}
			// when node folder is populated by FileServer, there should be three
			// tokens - path to task, partition id, partition content string
			else if(token.length == 4){
				
				System.out.println("Update invoked by FileServer.");
				String[] dictionaryWords = token[3].split(FileServer.DELIMITER);
				
				// calculate the md5sum and compare to the received password hash
				String match = null;
				boolean m = true;
				for(String dw: dictionaryWords){
					
					
					String dwHash = getHash(dw);
					
					if(m){
						System.out.println(dw + " is " + dwHash + " vs " + currentHash);
						m=false;
					}
					
					if(dwHash.equals(currentHash)){
						match = dw;
						break;
					}
				}
				
				if(match!= null){
					// write match to the job folder
					String jobFolder = currentTaskFolder.substring(0,currentTaskFolder.lastIndexOf("/"));
					System.out.println("Info: Match found. Writing [" + match+ "] to " + jobFolder);
					zkc.atomicAppend(match, jobFolder);
				}
				else{
					System.out.println("Info: No match found!");
				}
				
				// delete task folder
				try{
					System.out.println("Info: Deleting task folder " + currentTaskFolder);
					zkc.getZooKeeper().delete(currentTaskFolder, -1);
				}
				catch(KeeperException ke){
					System.out.println("Error: Cannot delete " + token[1] + " " + ke.getMessage());
				}
				catch(InterruptedException ie){
					System.out.println("Error: Cannot delete " + token[1] + " " + ie.getMessage());
				}
				
				// clear data in your folder
				System.out.println("Info: Clearing data in folder " + zNodePath);
//				zkc.atomicGetAndClearData(zNodePath);
				
				try{
					Stat dontcare = zkc.getZooKeeper().setData(zNodePath, "".getBytes(), -1);
					byte [] cb= zkc.getZooKeeper().getData(zNodePath, false, dontcare);
					
					if(cb == null || cb.length == 0){
						System.out.println("Info: Successfully cleared the content of folder");
					}
					else{
						System.out.println("Error: remaining content [" + new String(cb) +"]");
					}
				}
				catch (Exception e) {
					System.out.println("Error: Cannot clear the data inside this Worker's folder.");
				}
			}		
			
			attachWatcher();
			
		}
		
	}
	
	public void processWorkerFolderContent(String content, String passwordHash, String path ) {
		System.out.println("Update invoked by JobTracker.");
		currentTaskFolder = content;
		
		String taskFolderContent = null;
		
		try{
			// go in the task folder and retrieve the password hash and partition id
			byte[] taskFolderContentByte = zkc.getZooKeeper().getData(passwordHash, false, new Stat());
			taskFolderContent= new String(taskFolderContentByte);
			
		}catch(KeeperException ke){
			System.out.println("Error: Cannot read the data in " + path + " " + ke.getMessage());
			return;
		}
		catch(InterruptedException ie){
			System.out.println("Error: Cannot read the data in " +path + " " + ie.getMessage());
			return;
		}
		
		String[] taskToken = taskFolderContent.split(ZkConnector.DATADELIMITER);
		
		System.out.println("Info: taskFolder content " +taskFolderContent);
//		if(taskToken.length != 3){
//			System.out.println("Error: Malformed information contained in the task folder [" + taskFolderContent + "]");
//			attachWatcher();
//			return;
//		}
						
		System.out.println("Info: Creating a packet for FileServer partition request.");
		
		currentHash = taskToken[0];
		String partitionRequest = zNodePath + ZkConnector.DATADELIMITER + taskToken[1];
			
		// now create a File request for partition retrivel to FileServer 
		Code ret = zkc.create(FILE_REQUEST_PATH, partitionRequest, CreateMode.EPHEMERAL);
//		
//		if(ret != KeeperException.Code.OK){
//			
//		}

		System.out.println("Sent [" + partitionRequest + "] data to " + FILE_REQUEST_PATH);
		
	}

	/**
	 * Attaches the personal Worker watcher that will see the changes in this zNode.
	 */
	private void attachWatcher(){
		Stat stat = zkc.exists(zNodePath, workerWatcher);

		// zNode doesn't exist; let's try creating it
		if (stat == null) {
			System.out.println("Error: This block should never be reached. Eitherway, process is exiting...");
			System.exit(-1);
		}
		
		System.out.println("Info: Successfully attached watcher.");
	}
	
	private static String getHash(String word) {

        String hash = null;
        try {
            MessageDigest md5 = MessageDigest.getInstance("MD5");
            BigInteger hashint = new BigInteger(1, md5.digest(word.getBytes()));
            hash = hashint.toString(16);
            while (hash.length() < 32) hash = "0" + hash;
        } catch (NoSuchAlgorithmException nsae) {
            // ignore
        }
        return hash;
    }
	
	public ZooKeeper getZooKeeper(){
		return zkc.getZooKeeper();
	}
	
	public String getMyPath(){
		return zNodePath;
	}
	
	/**
	 * Starts the process.
	 * 
	 * @param args	command line arguments
	 */
	public static void main(String[] args) {

		System.out.println("Worker starting now");

		if (args.length != 1) {
			System.out
					.println("Usage: java -classpath lib/zookeeper-3.3.2.jar:lib/log4j-1.2.15.jar:. Worker zkServer:clientPort");
			return;
		}

		Worker worker = new Worker(args[0]);
		worker.attachWatcher();

		System.out.println("Spin...");
		
		while (true) {
			
			String content = worker.print();
			System.out.println("Current worker contains [" +content+ "]");
			String path = worker.getMyPath();
			
//			
//			try{
//				Stat readByteStat = new Stat();
//				byte[] readByte = worker.getZooKeeper().getData(path, true, readByteStat);
//				
//				content = new String(readByte);
//				
//				System.out.println("Info: Read [" + content + "] from " +path);
//								
//				if(content.isEmpty()){
//					System.out.println("Error: The data received in "+ path+ " is empty!");
//					continue;
//				}
//			}
//			catch(KeeperException ke){
//				System.out.println("Error: Cannot read the data in " + path + " " + ke.getMessage());
//				continue;
//			}
//			catch(InterruptedException ie){
//				
//				System.out.println("Error: Cannot read the data in " + path+ " " + ie.getMessage());
//				continue;
//			}
			
			// parse the content of the data inside
			String[] token = content.split(ZkConnector.DATADELIMITER);

			System.out.println("Info: Received " + token.length + " tokens.");
			for(String t: token){
				System.out.println("===================");
				System.out.println(t);
			}
			
			// when invoked by JobTracker, there should only be one token in the 
			// data received - path to task. 
			if(content.length() >0 && token.length == 1){

				worker.processWorkerFolderContent(content, token[0], path);
				
			}
			
			try {
				Thread.sleep(SLEEP_TIME);
			} catch (Exception e) {
			}
			// spin!
		}
	}
	
	 
    private String print() {
    	Stat getDataStat = new Stat();
    	
    	try{
		byte[] updatedDataByte = zkc.getZooKeeper().getData(zNodePath,
				false, getDataStat); 
		
		if(updatedDataByte != null && updatedDataByte.length > 0 ){
		
			 return new String(updatedDataByte);
			}
    	}
    	catch(Exception e){
    		
    	}
    	
    	return "";
		
	}


	/** Processing interval for testing */
    private static final int PROCESSING_INTERVAL = 4*SLEEP_TIME;

    /** Mocked password hash - d00rst0p */
    private static final String MOCK_HASH = getHash("d00rst0p");
    
    /** Initial mocked data */
    private static final String MOCK_FOLDER_INITIAL_DATA_TEST =  MOCK_HASH + ZkConnector.DATADELIMITER + "0";
    
    /** Mocked job folder path */
    private static final String MOCK_JOB_FOLDER = PoolCreator.JOBPOOL_PATH + "/" + MOCK_HASH;
    
    /** Initialization flag */
    private boolean mockFlagInitialized = false;
    
    /** Watcher for job folder */
    private Watcher mockWatcher;
    
    /** Path to the mocked task folder */
    private String mockTaskPath;
    
	/**
	 * Populates the zNode of this Worker
	 */
	private void fireMock(String zkhost){
	
		
		
		if(!mockFlagInitialized){
					
			refreshMock();
			PoolCreator.createPool(zkc, PoolCreator.JOBPOOL_PATH);
			
			System.out.println("[MOCK] Info: Initializing mocked path.");

			Code ret = zkc.create(MOCK_JOB_FOLDER, null, CreateMode.PERSISTENT);
					
			if(ret != Code.OK){
				System.out.println("[MOCK] Error: Cannot initialize " + MOCK_JOB_FOLDER);
				cleanMock();
			}
			
			mockTaskPath = MOCK_JOB_FOLDER + "/task";
			
			try{
				mockTaskPath = zkc.getZooKeeper().create(mockTaskPath, MOCK_FOLDER_INITIAL_DATA_TEST.getBytes(), zkc.getACL(), CreateMode.EPHEMERAL_SEQUENTIAL);
				System.out.println("[MOCK] Info: Created new task path " + mockTaskPath);			
			}
			catch(KeeperException ke){
				System.out.println("[MOCK] Error: Cannot initialize " + mockTaskPath + " - " + ke.getMessage());
				cleanMock();
			}
			catch(InterruptedException ie){
				System.out.println("[MOCK] Error: Cannot initialize " + mockTaskPath + " - " + ie.getMessage());
				cleanMock();
			}

			mockWatcher = new Watcher(){
				
				//@Override
				public void process(WatchedEvent event) {
					
					if(event.equals(EventType.NodeDataChanged)){
						
						try{
							Stat getDataStat = new Stat();
							byte[] updatedDataByte = zkc.getZooKeeper().getData(event.getPath(),
									false, getDataStat); 
							
							String updatedData = new String(updatedDataByte);
							
							System.out.println("[MOCK] Info: Mocked task path updated data to [" + updatedData + "]");
						}
						catch(KeeperException ke){
							System.out.println("[MOCK] Exception encountered in mocked zNode update " + ke.getMessage());
						}
						catch(InterruptedException ie){
							System.out.println("[MOCK] Exception encountered in mocked zNode update " + ie.getMessage());
						}
					}
				}
			};
			
			// attach the watcher to job folder so events triggered by Worker can be captured.
			Stat stat = zkc.exists(MOCK_JOB_FOLDER, mockWatcher);

			if (stat == null) {
				System.out.println("[MOCK] Error: This block should never be reached. Eitherway, process is exiting...");
				cleanMock();
			}
			
			 
			mockFlagInitialized = true;
			
			System.out.println("[MOCK] Info: Successfully initialized " + mockTaskPath);
		}
		
		// write data in your own folder
		zkc.atomicAppend(mockTaskPath, zNodePath);

		try {
			Thread.sleep(PROCESSING_INTERVAL);			
		} catch (Exception e) {
		}
	}

	/**
	 * Refreshes the mocked zNode paths.
	 */
	private void refreshMock(){
		System.out.println("Cleaning mocked setup.");
		try{
			Stat s = zkc.exists(mockTaskPath, false);
		
			if(s!=null){
				
				System.out.println("Deleting " + mockTaskPath);
				zkc.getZooKeeper().delete(mockTaskPath, -1);
			}
		}
		catch(KeeperException ke){
			System.out.println("[MOCK] Warning: Exception when deleting " + mockTaskPath + " " + ke.getMessage());
		}
		catch(InterruptedException ie){
			System.out.println("[MOCK] Warning: Exception when deleting " + mockTaskPath + " " + ie.getMessage());
		}
		
		try{
			Stat s = zkc.exists(MOCK_JOB_FOLDER, false);
		
			if(s!=null){
				System.out.println("Deleting " + MOCK_JOB_FOLDER);
				zkc.getZooKeeper().delete(MOCK_JOB_FOLDER, -1);
			}
		}
		catch(KeeperException ke){
			System.out.println("[MOCK] Warning: Exception when deleting " + MOCK_JOB_FOLDER + " " + ke.getMessage());
		}
		catch(InterruptedException ie){
			System.out.println("[MOCK] Warning: Exception when deleting " + MOCK_JOB_FOLDER + " " + ie.getMessage());
		}
		
		try{
			Stat s = zkc.exists(PoolCreator.JOBPOOL_PATH, false);
		
			if(s!=null){
				System.out.println("Deleting " + PoolCreator.JOBPOOL_PATH);
				zkc.getZooKeeper().delete(PoolCreator.JOBPOOL_PATH, -1);
			}
		}
		catch(KeeperException ke){
			System.out.println("[MOCK] Warning: Exception when deleting " + PoolCreator.JOBPOOL_PATH + " " + ke.getMessage());
		}
		catch(InterruptedException ie){
			System.out.println("[MOCK] Warning: Exception when deleting " + PoolCreator.JOBPOOL_PATH + " " + ie.getMessage());
		}
	}
	/** 
	 * Cleans up the folders created by mocked case. 
	 */
	private void cleanMock() {
		
		refreshMock();		
		System.exit(-1);
		
	}
	
	
}
